package com.relytosoft.mizar.media.system.services.consumer;

import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.relytosoft.mizar.media.external.api.system.request.DistrictDto;
import com.relytosoft.mizar.media.external.api.system.response.AddRegionDto;
import com.relytosoft.mizar.media.external.api.system.response.QueryRegionDto;
import com.relytosoft.mizar.media.external.api.system.response.RegionDto;
import com.relytosoft.mizar.media.external.api.system.service.RegionService;
import com.relytosoft.mizar.media.system.api.entity.SysDistrict;
import com.relytosoft.mizar.media.system.api.entity.SysRelatedPlatform;
import com.relytosoft.mizar.media.system.services.service.SysRelatedPlatformService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * 区域消息消费
 *
 * @Author SunLi
 * @Date: 2022/4/13 9:43
 * @Version 1.0
 */
@Component
public class DistrictConsumer {

    protected Logger log = LoggerFactory.getLogger(this.getClass());

//    @Resource
//    private KafkaProducer kafkaProducer;

    @Resource
    private RegionService regionService;

    @Resource
    private SysRelatedPlatformService sysRelatedPlatformService;

//    @KafkaListener(id = "consumer-media-region-sync", topics = MediaMqConstants.REGION_TOPIC, containerFactory = "singleContainerFactory")
//    public void processSingleMessage(ConsumerRecord<?, ?> record, Acknowledgment ack) {
//        try {
//            boolean isConsume = false;
//            KafkaOpMsg kafkaOpMsg = JSON.parseObject(record.value().toString(), KafkaOpMsg.class);
//            switch (kafkaOpMsg.getOpType()) {
//                //查询第三方根区域
//                case "QUERYROOT":
//                    isConsume = queryRoot();
//                    break;
//                //添加根区域至系统
//                case "INSERTROOT":
//                    isConsume = insertRoot(kafkaOpMsg.getData());
//                    break;
//                //添加区域至第三方
//                case "INSERT":
//                    isConsume = insert(kafkaOpMsg.getData());
//                    break;
//                //更新区域关联信息至系统
//                case "UPDATERELEATED":
//                    isConsume = updateRelated(kafkaOpMsg.getData());
//                    break;
//                //查询第三方区域信息
//                case "QUERY":
//                    isConsume = queryRegion(kafkaOpMsg.getData());
//                    break;
//                default:
//            }
//            log.info("线程:{} 消费->{}", Thread.currentThread(), kafkaOpMsg.getOpType());
//            if (isConsume) {
//                ack.acknowledge();
//            }
//        } catch (Exception e) {
//            log.error(e.getMessage());
//        }
//    }

    /**
     * 查询第三方根区域，并发送添加根区域消息
     *
     * @return 消费成功
     */
    private boolean queryRoot() {
        //查询海康根区域
        RegionDto regionDTO = regionService.getRootRegion();
        if (regionDTO != null) {
            //发送添加根区域消息
//            KafkaOpMsg kafkaOpMsg = new KafkaOpMsg();
//            kafkaOpMsg.setOpType("INSERTROOT");
//            kafkaOpMsg.setData(JSON.toJSONString(regionDTO));
//            kafkaProducer.send(MediaMqConstants.REGION_TOPIC, JSON.toJSONString(kafkaOpMsg), true);
            return true;
        }
        return false;
    }

    /**
     * 添加根区域信息
     *
     * @param data 消息信息
     * @return 消费结果
     */
    private boolean insertRoot(String data) {
        //查询系统根区域信息
        SysRelatedPlatform sysRelatedPlatform = sysRelatedPlatformService.getRelatedPlatform(1, "HK", 0, 1);

        if (sysRelatedPlatform == null) {
            //不存在则添加
            RegionDto regionDTO = JSON.parseObject(data, RegionDto.class);
            sysRelatedPlatform = new SysRelatedPlatform();
            sysRelatedPlatform.setOwnId(0);
            sysRelatedPlatform.setType(1);
            sysRelatedPlatform.setPlatformName("HK");
            sysRelatedPlatform.setPlatformId(regionDTO.getIndexCode());
            sysRelatedPlatform.setIsSync(1);
            sysRelatedPlatform.setStatus(1);
            return sysRelatedPlatformService.save(sysRelatedPlatform);
        } else {
            //存在直接返回已消费
            return true;
        }
    }

    /**
     * 添加区域信息至第三方
     *
     * @param data 消息信息
     * @return 消费结果
     */
    private boolean insert(String data) {
        SysDistrict entity = JSON.parseObject(data, SysDistrict.class);
        //父级区域关联第三方标识为空
        if (StrUtil.isEmpty(entity.getParentPlatformId())) {
            //查询系统父级区域关联第三方
            SysRelatedPlatform sysRelatedPlatform = sysRelatedPlatformService.getRelatedPlatform(1, "HK", entity.getParentId(), 1);
            entity.setParentPlatformId(sysRelatedPlatform.getPlatformId());
        }
        if (!StrUtil.isEmpty(entity.getParentPlatformId())) {
            //父级区域已关联至第三方
            //调用第三方添加区域
            AddRegionDto addRegionDTO = regionService.addRegion(new DistrictDto(entity.getName(), entity.getRemark(), entity.getParentPlatformId()));
            if (addRegionDTO.getSuccesses() != null && addRegionDTO.getSuccesses().size() > 0) {
                //第三方添加区域成功
                //发送更新第三方区域信息到关联表的消息
                SysRelatedPlatform curRelatedPlatform = new SysRelatedPlatform();
                curRelatedPlatform.setId(entity.getRelatedPlatformId());
                curRelatedPlatform.setIsSync(1);
                curRelatedPlatform.setPlatformId(addRegionDTO.getSuccesses().get(0).getRegionIndexCode());
                curRelatedPlatform.setUpdatedBy(entity.getUpdatedBy());
                curRelatedPlatform.setUpdatedId(entity.getUpdatedId());
//                KafkaOpMsg kafkaOpMsg = new KafkaOpMsg();
//                kafkaOpMsg.setOpType("UPDATERELEATED");
//                kafkaOpMsg.setData(JSON.toJSONString(curRelatedPlatform));
//                kafkaProducer.send(MediaMqConstants.REGION_TOPIC, JSON.toJSONString(kafkaOpMsg), true);
                return true;
            } else {
                //第三方添加区域失败，发送查询区域消息
//                KafkaOpMsg kafkaOpMsg = new KafkaOpMsg();
//                kafkaOpMsg.setOpType("QUERY");
//                kafkaOpMsg.setData(JSON.toJSONString(entity));
//                kafkaProducer.send(MediaMqConstants.REGION_TOPIC, JSON.toJSONString(kafkaOpMsg), true);
                return true;
            }
        }
        return false;
    }

    /**
     * 更新第三方关联信息
     *
     * @param data 消息信息
     * @return 消费结果
     */
    private boolean updateRelated(String data) {
        SysRelatedPlatform curRelatedPlatform = JSON.parseObject(data, SysRelatedPlatform.class);
        return sysRelatedPlatformService.updateById(curRelatedPlatform);
    }

    /**
     * 查询第三方区域信息
     *
     * @param data 消息信息
     * @return 消费结果
     */
    private boolean queryRegion(String data) {
        SysDistrict entity = JSON.parseObject(data, SysDistrict.class);
        //查询第三方区域信息
        QueryRegionDto queryRegionDTO = regionService.searchRegion(new DistrictDto(entity.getName(), "", entity.getParentPlatformId()));

        if (queryRegionDTO.getTotal() > 0) {
            //更新第三方区域信息到关联表
            SysRelatedPlatform curRelatedPlatform = new SysRelatedPlatform();
            curRelatedPlatform.setId(entity.getRelatedPlatformId());
            curRelatedPlatform.setIsSync(1);
            curRelatedPlatform.setPlatformId(queryRegionDTO.getList().get(0).getIndexCode());
            curRelatedPlatform.setUpdatedBy(entity.getUpdatedBy());
            curRelatedPlatform.setUpdatedId(entity.getUpdatedId());

//            KafkaOpMsg kafkaOpMsg = new KafkaOpMsg();
//            kafkaOpMsg.setOpType("UPDATERELEATED");
//            kafkaOpMsg.setData(JSON.toJSONString(curRelatedPlatform));
//            kafkaProducer.send(MediaMqConstants.REGION_TOPIC, JSON.toJSONString(kafkaOpMsg), true);
        }
        return true;
    }
}
